use std::{collections::BTreeMap, time::Duration};

use futures::TryStreamExt;
use pulsar::{Consumer, Pulsar, SubType, TokioExecutor};
use regex::Regex;
use tokio::sync::{
  mpsc::{self, error::TryRecvError},
  oneshot,
};
use tracing::{debug, error, warn};

const CALL_ID_REG_STR: &str = r##""_call_id" *: *(-?\d+) *,"##;

///注册一个返回通知消息
pub(crate) struct CallResultRegister {
  pub call_id: i32,
  pub sender: oneshot::Sender<String>,
}

/// 调用方法返回消息的接收器
///
/// 此接收器依赖消息队列工作.
/// 此接收器需要指定对应的主题名和sub名称进行接收
/// 在此接收器第一次启动的时候将返回一个发送器,此发送器对应的接收者将用来注册需要收到的返回对象.
/// 当此对象收到一个消息时,将从返回信息的队列当中查找对应的接收器,并将消息发送给对应的接收器.
/// 如果不存在,放弃此消息.如果发送失败,跳过此消息.
pub(crate) struct CallBackReceiver {
  ///等待收到返回信息的队列
  wait_call_backs: BTreeMap<i32, oneshot::Sender<String>>,
  ///消费者,用来接收消息队列传回来的返回值数据
  consumer: Consumer<String, TokioExecutor>,
  call_receiver: mpsc::UnboundedReceiver<CallResultRegister>,
  call_id_reg: Regex,
}

impl CallBackReceiver {
  pub async fn start_running(
    topic_name: String,
    sub_name: String,
    pulsar: &Pulsar<TokioExecutor>,
  ) -> Result<mpsc::UnboundedSender<CallResultRegister>, pulsar::Error> {
    let consumer = pulsar
      .consumer()
      .with_topic(topic_name)
      .with_consumer_name("load_call_return")
      .with_subscription_type(SubType::Exclusive)
      .with_subscription(sub_name)
      .build()
      .await?;

    let (call_id_sender, call_id_receiver) = mpsc::unbounded_channel();

    let receiver = CallBackReceiver {
      wait_call_backs: BTreeMap::new(),
      consumer,
      call_receiver: call_id_receiver,
      call_id_reg: Regex::new(CALL_ID_REG_STR).expect("初始call_id_reg的化正则表达式失败"),
    };

    tokio::spawn(receiver.run());

    Ok(call_id_sender)
  }

  async fn run(mut self) {
    loop {
      //先接收来自本地的需要得到返回值的注册信息
      match self.call_receiver.try_recv() {
        Ok(reg) => {
          self.wait_call_backs.insert(reg.call_id, reg.sender);
          //加入成功后,跳过后续执行,目的是处理完所有注册信息再去处理后续
          continue;
        }
        Err(TryRecvError::Empty) => {}
        Err(TryRecvError::Disconnected) => {
          error!("方法返回注册发送通道断开,无法再收到注册信息");
        }
      }

      //不存在需要返回值的消息,则跳过消息队列来的数据
      if self.wait_call_backs.is_empty() {
        //队列为空,则等待100毫秒
        tokio::time::sleep(Duration::from_millis(100)).await;
        continue;
      }

      //只获取一次消息队列的值,最大程度保证注册早于消息队列来到
      let msg = match self.consumer.try_next().await {
        Ok(Some(msg)) => msg,
        Ok(None) => {
          warn!("从方法返回消息队列收到了一个 Ok(None)..是代表已经退出了?");
          continue;
        }
        Err(e) => {
          error!("获取消息队列数据失败{}", e);
          continue;
        }
      };

      if let Err(e) = self.consumer.ack(&msg).await {
        error!("从方法返回确认消息队列数据失败{}", e);
      }

      let msg_string = match msg.deserialize() {
        Ok(m) => m,
        Err(e) => {
          error!("从方法返回消息队列数据反序列化失败{}", e);
          continue;
        }
      };

      let Some(call_id) = self.load_call_id(&msg_string) else {
        error!("从方法返回消息队列数据解析call_id失败");
        continue;
      };

      //找到对应的发送器,删除并发送
      match self.wait_call_backs.remove(&call_id) {
        Some(sender) => {
          if let Err(e) = sender.send(msg_string) {
            debug!("从方法返回消息队列数据发送失败..待发送的消息:{}", e);
          }
        }
        None => {
          debug!("从方法返回收到一个未注册的call_id,收到的消息:{msg_string}");
          continue;
        }
      }
    }
  }

  #[inline]
  fn load_call_id(&self, msg: &str) -> Option<i32> {
    match self
      .call_id_reg
      .captures(msg)
      .and_then(|cap| cap.get(1).map(|m| m.as_str().parse::<i32>()))
    {
      Some(Ok(r)) => Some(r),
      Some(Err(e)) => {
        error!("解析call_id失败,错误:{e}");
        None
      }
      None => {
        error!("没有找到call_id");
        None
      }
    }
  }
}
